package fun.tan90.easy.binlog.plugin.kafka.service.impl;

import cn.hutool.json.JSONUtil;
import fun.tan90.easy.binlog.plugin.common.service.PluginService;
import fun.tan90.easy.binlog.plugin.kafka.model.KafkaClientParams;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Service;

import java.util.Collections;
import java.util.Properties;

/**
 * description TODO
 * company 铁人科技
 *
 * @author 陈浩杰
 * @date 2023/11/15 14:41
 */
@Slf4j
@Service(value = "kafka")
public class KafkaPluginService implements PluginService {
    @Override
    public boolean test(String params) {
        KafkaClientParams kafkaClientParams = JSONUtil.toBean(params, KafkaClientParams.class);
        Properties props = new Properties();
        /**
         * bootstrap.servers（必需）：指定 Kafka 集群的地址列表，格式为 "host1:port1,host2:port2,..."。生产者将使用这个参数来连接到 Kafka 集群。例如：bootstrap.servers=localhost:9092。
         * key.serializer 和 value.serializer（必需）：指定键（key）和值（value）的序列化器类。生产者需要将消息中的键和值序列化为字节流以进行传输。这两个参数的值通常设置为实现了 org.apache.kafka.common.serialization.Serializer 接口的序列化器类。例如：key.serializer=org.apache.kafka.common.serialization.StringSerializer。
         * acks：指定生产者等待服务器确认消息写入成功的方式。可以设置的值包括 "all"（需要所有副本确认），"1"（仅需要 Leader 副本确认），"0"（不需要确认，默认值）。使用 "all" 可以提供最高的消息可靠性，但会降低生产者的吞吐量。
         * retries：指定生产者在消息发送失败时的重试次数。当发生可重试错误时，生产者将尝试重新发送消息。默认值为 0，表示不进行重试。
         * batch.size：指定生产者在发送到同一个分区之前等待累积的消息批次大小（字节）。默认值为 16384（16KB）。较大的批次大小可以提高吞吐量，但也会增加延迟。
         * linger.ms：指定生产者在发送消息之前等待的最长时间（毫秒）。默认值为 0，表示生产者将立即发送消息而不进行等待。较大的值可以增加消息的批次大小，从而提高吞吐量，但会增加一定的延迟。
         */
        String bootstrapServers = kafkaClientParams.getBootstrapServers();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.ACKS_CONFIG, kafkaClientParams.getAcks());
        props.put(ProducerConfig.RETRIES_CONFIG, kafkaClientParams.getRetries());
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, kafkaClientParams.getBatchSize());
        props.put(ProducerConfig.LINGER_MS_CONFIG, kafkaClientParams.getLingerMs());
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, kafkaClientParams.getBufferMemory());
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, kafkaClientParams.getCompressionType());
        props.put(ProducerConfig.METADATA_MAX_AGE_CONFIG, kafkaClientParams.getMetadataMaxAgeMs());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // 测试用
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 3000);
        String topic = "ping";

        try (KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props)) {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, "ping");
            kafkaProducer.send(record).get();

            props.clear();

            props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            try (AdminClient adminClient = AdminClient.create(props)) {
                adminClient.deleteTopics(Collections.singleton(topic));
            }

            return true;
        } catch (Exception e) {
            e.printStackTrace();
            log.error("kafka 连接测试失败，{}", e.getMessage());
            return false;
        }
    }
}
